package com.zhouxiaoge;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author 周小哥 gqzmy@163.com 2022/7/21
 */
public class ClickSource implements SourceFunction<Event> {

    private final AtomicBoolean atomicBoolean = new AtomicBoolean(true);

    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        String[] users = {"Bob", "Tom", "Jerry", "Alice"};
        String[] urls = {"/home", "/cart", "/pay", "/order?id=111", "/order?id=222"};
        Random random = new Random();
        while (this.atomicBoolean.get()) {
            String user = users[random.nextInt(users.length)];
            String url = urls[random.nextInt(users.length)];
            ctx.collect(new Event(user, url, Calendar.getInstance().getTimeInMillis()));
            Thread.sleep(random.nextInt(3) * 1000);
        }
    }

    @Override
    public void cancel() {
        this.atomicBoolean.set(false);
    }
}
